package job;

import bean.TrafficEvent;
import function.TrafficMapFunction;
import net.bwie.realtime.traffic.common.utils.JdbcUtil;
import net.bwie.realtime.traffic.common.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author 26374
 */
public class TransportationJob1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<String> kafka = KafkaUtil.consumerKafka(env, "traffic_topic");
        DataStream<String> dataStream = handle(kafka);

        dataStream.print();
        JdbcUtil.sinkToClickhouseUpsert(dataStream,
                "insert into traffic_monitoring.dws_city_overview(\n" +
                        "    totalFlow, avgSpeed, congestionRate, congestionLevel, ts)\n" +
                        "VALUES (?,?,?,?,?)");
        env.execute("TransportationJob");
    }

    private static DataStream<String> handle(DataStream<String> kafka) {
        DataStream<TrafficEvent> watermarks = kafka
                .map(new TrafficMapFunction())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, ts) -> event.getTimestamp()));

        // 全局窗口（不按 roadId 分组）
        DataStream<String> metrics = watermarks
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new ProcessAllWindowFunction<TrafficEvent, String, TimeWindow>() {
                    final double FREE_FLOW_SPEED = 80.0;
                    int totalFlow = 0;
                    double totalSpeed = 0.0;
                    int congested = 0;
                    @Override
                    public void process(
                            Context context,
                            Iterable<TrafficEvent> events,
                            Collector<String> out) {

                        // 计算总车流量


                        Long timestamp = 0L;
                        for (TrafficEvent event : events) {
                            totalFlow++;
                            totalSpeed += event.getSpeed();
                            if (event.getSpeed() < FREE_FLOW_SPEED * 0.5) {
                                congested++;
                            }
                            timestamp = event.getTimestamp();
                        }

                        // 计算平均车速和拥堵率
                        double avgSpeed = totalFlow == 0 ? 0 : totalSpeed / totalFlow;
                        double congestionRate = totalFlow == 0 ? 0 : (double) congested / totalFlow * 100;
                        String level = congestionRate > 60 ? "严重拥堵" :
                                congestionRate > 30 ? "中度拥堵" : "畅通";

                        // 输出汇总结果
//                        TrafficMetrics result = new TrafficMetrics();
//                        result.setTotalFlow(totalFlow);
//                        result.setAvgSpeed(Math.round(avgSpeed * 100) / 100.0);
//                        result.setCongestionRate(Math.round(congestionRate * 100) / 100.0);
//                        result.setCongestionLevel(level);
//                        result.setTimestamp(timestamp);
                        String ss = totalFlow  + "," + Math.round(avgSpeed * 100) / 100.0 + "," + Math.round(congestionRate * 100) / 100.0 + "," + level + "," + timestamp;
                        out.collect(ss);
                    }
                });

        // 打印结果
        return metrics;
    }

}
